package com.fanshuai.spring.boot.component;

import com.fanshuai.spring.boot.component.handler.MessageHandler;
import com.fanshuai.spring.boot.component.listener.ConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 通用的kafka listener, 通用自定义handler处理
 */
@Slf4j
public class GenericKafkaListener implements AcknowledgingMessageListener<String, String> {
    private List<MessageHandler> handlerList;
    private List<ConsumerListener> consumerListeners;
    private ConsumerFactory<String, String> consumerFactory;

    public GenericKafkaListener(List<MessageHandler> handlerList, List<ConsumerListener> consumerListeners, ConsumerFactory<String, String> factory) {
        this.handlerList = handlerList;
        this.consumerListeners = consumerListeners;
        this.consumerFactory = factory;
    }

    public void onMessage(ConsumerRecord<String, String> data, Acknowledgment ack) {
        if (CollectionUtils.isEmpty(handlerList)) {
            return;
        }

        List<ConsumerListener> listeners = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(consumerListeners)) {
            listeners = consumerListeners.stream().sorted(Comparator.comparingInt(ConsumerListener::getOrder))
                    .collect(Collectors.toList());

            for (ConsumerListener listener : listeners) {
                try {
                    listener.beforeConsume(data);
                } catch (Exception e) {
                    log.error("listen beforeConsume error, listener={}, ex={}", listener.getClass().getName(), e);
                }
            }
        }

        String topic = data.topic();
        for (MessageHandler handler : handlerList) {
            if (handler.topics().contains(topic)) {
                try {
                    handler.handle(topic, data.value());

                    if (CollectionUtils.isNotEmpty(listeners)) {
                        for (ConsumerListener listener : listeners) {
                            try {
                                listener.afterConsume(data);
                            } catch (Exception e) {
                                log.error("listen afterConsume error, listener={}, ex={}", listener.getClass().getName(), e);
                            }
                        }
                    }

                    if (!consumerFactory.isAutoCommit()) {
                        ack.acknowledge();
                    }
                } catch (Exception e) {
                    log.error("handle kafka message error, message={}", data);
                }

                return;
            }
        }
    }
}
